package rpc

import (
	"bytes"
	"cluster-cache/common"
	"encoding/binary"
	"errors"
	"github.com/panjf2000/gnet/pkg/logging"
	"io"
	"sync"
)

// HeadBodyCodec Protocol format:
//
// one connection one codec
// * 0           2  --(8)--  10       --(4)--       14
// * +-----------+-----------+-----------------------+
// * |   magic   |    seq    |       body len        |
// * +-----------+-----------+-----------------------+
// * |                                               |
// * +                                               +
// * |                 body bytes                    |
// * +                                               +
// * |                  ... ...                      |
// * +-----------------------------------------------+

var ErrIncompletePacket = errors.New("incomplete packet")
var ErrInvalidMagicNumber = errors.New("invalid magic number")

const (
	magicNumber     = 1314
	magicNumberSize = 2
	seqNumberSize   = 8
	bodySize        = 4
	seqOffset       = magicNumberSize + seqNumberSize // 10
	bodyOffset      = seqOffset + bodySize            // 14
)

var magicNumberBytes []byte

func init() {
	magicNumberBytes = make([]byte, magicNumberSize)
	binary.BigEndian.PutUint16(magicNumberBytes, uint16(magicNumber))
}

type HeadBodyCodec struct {
	finished bool
	head     *bytes.Buffer
	body     *bytes.Buffer
	bodyLen  int
	headTmp  []byte
	pos      uint64
	Seq      *common.AsyncSeq
}

var HeadBodyCodecPool = sync.Pool{New: func() interface{} {
	return NewHeadBodyCodec()
}}

func NewHeadBodyCodec() *HeadBodyCodec {
	return &HeadBodyCodec{
		head:     new(bytes.Buffer),
		body:     new(bytes.Buffer),
		finished: true,
		bodyLen:  -1,
		headTmp:  make([]byte, bodyOffset),
		pos:      0,
		Seq:      common.NewAtomicSeq(),
	}
}

func (h *HeadBodyCodec) Decode(c io.Reader) ([]byte, uint64, error) {
	if h.finished {
		x, err := c.Read(h.headTmp[0:(bodyOffset - h.head.Len())])
		if err != nil {
			return nil, 0, err
		}
		h.head.Write(h.headTmp[:x])
		if h.head.Len() < bodyOffset {
			return nil, 0, ErrIncompletePacket
		} else if h.head.Len() == bodyOffset {
			if !bytes.Equal(magicNumberBytes, h.head.Bytes()[:magicNumberSize]) {
				return nil, 0, ErrInvalidMagicNumber
			}
			h.pos = binary.BigEndian.Uint64(h.head.Bytes()[magicNumberSize:seqOffset])
			h.bodyLen = int(binary.BigEndian.Uint32(h.head.Bytes()[seqOffset:bodyOffset]))
			h.finished = false
		} else {
			logging.Errorf("head=%v, body=%v", h.head.Bytes(), h.body.Bytes())
			panic("decode panic")
		}
	}
	for h.body.Len() < h.bodyLen {
		other := make([]byte, h.bodyLen-h.body.Len())
		x, err := c.Read(other)
		if x == 0 {
			break
		}
		if err != nil {
			return nil, 0, err
		}
		h.body.Write(other[:x])
	}
	if h.body.Len() < h.bodyLen {
		h.finished = false
		return nil, 0, ErrIncompletePacket
	} else if h.body.Len() == h.bodyLen {
		h.finished = true
		return h.body.Bytes(), h.pos, nil
	} else {
		logging.Errorf("head=%v, body=%v", h.head, h.body)
		panic("decode panic")
	}
	// body may be empty
	return nil, 0, ErrIncompletePacket
}

func (h *HeadBodyCodec) Encode(buf []byte) ([]byte, uint64, error) {
	if buf == nil {
		return nil, 0, nil
	}
	seq := h.Seq.Increase()
	msgLen := bodyOffset + len(buf)
	data := make([]byte, msgLen)
	copy(data, magicNumberBytes)
	binary.BigEndian.PutUint64(data[magicNumberSize:seqOffset], seq)
	binary.BigEndian.PutUint32(data[seqOffset:bodyOffset], uint32(len(buf)))
	if msgLen == bodyOffset {
		return data, seq, nil
	}
	copy(data[bodyOffset:msgLen], buf)
	return data, seq, nil
}

func EncodeWithSeq(buf []byte, seq uint64) ([]byte, error) {
	if buf == nil {
		return nil, nil
	}
	msgLen := bodyOffset + len(buf)
	data := make([]byte, msgLen)
	copy(data, magicNumberBytes)
	binary.BigEndian.PutUint64(data[magicNumberSize:seqOffset], seq)
	binary.BigEndian.PutUint32(data[seqOffset:bodyOffset], uint32(len(buf)))
	if msgLen == bodyOffset {
		return data, nil
	}
	copy(data[bodyOffset:msgLen], buf)
	return data, nil
}

func (h *HeadBodyCodec) Reset() {
	h.finished = true
	h.head.Reset()
	h.body.Reset()
	h.bodyLen = -1
}
